package day05

import java.sql.{Date, DriverManager}

import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}

object JdbcRDDDemo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("jdbcrdddemo").setMaster("local[2]")
    val sc = new SparkContext(conf)

    // 下面链接数据库函数DriverManager.getConnection(url,user,pwd)需要指定三个参数：数据库url、数据库用户名、用户名密码
    /**
      * url为："jdbc:mysql://192.168.161.1:3306/bigdata?useUnicode=true&characterEncoding=utf8"
      * 格式：
      *   前面的jdbc:代表的是通过jdbc获取
      *   后面跟mysql的url 即  mysql://192.168.88.83:3306/
      *   在接着跟数据库名
      *   在接着跟 ?编码   即 ?useUnicode=true&characterEncoding=utf8
      */
    val url = "jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=utf8"
    val user = "root"
    val pwd = "123456"

    // JdbcRDD需要一个数据库连接的函数，这里创建该函数
    val conn = () => {
      //通过反射机制拿到驱动类，这里请求Mysql,还需要newInstance得到实例
      Class.forName("com.mysql.jdbc.Driver").newInstance()
      DriverManager.getConnection(url,user,pwd)
    }

    // JdbcRDD(sc,数据库连接的函数，sql查询语句字符串，查询开始点，查询结束点，分区数，处理返回数据的函数)
    // 注意这个sql查询语句参数格式，必须有where id >= ? and <=?因为这两个?要由后面的参数查询开始点和结束点参数指定
    val sql = "select * from location_info where id >= ? and id <= ?"
    val result: JdbcRDD[(Int, String, Int, Date)] = new JdbcRDD(
      sc, conn, sql, 0, 1000, 1,
      res => { //这个res是返回值，下面的操作这里我没处理数据，只是简单获取然后返回。
        val id = res.getInt("id")
        //getInt为得到该表id字段，因为id字段是int类型所以为getInt
        val location = res.getString("location")
        val counts = res.getInt("counts")
        val access_date = res.getDate("access_date")
        //返回处理后的数据
        (id, location, counts, access_date)
      }
    )

    println(result.collect().toBuffer)
  }
}
